
#include "BUtils.h"

using namespace blueberry::utl::que;

/*class BQueueContainer
 {
 public:
 BQueueContainer( void *);//Params - pointer to relevant data item
 BQueueContainer();
 void * p_Data;
 BQueueContainer* p_Next;
 };
 class IRSQueue
 {
 public:
 BQueueContainer * p_Head, * p_Tail;
 IRSQueue();//Creates an empty queue
 void AddtoQueue( BQueueContainer *);//Params - pointer to the BQueueContainer which should be added to the queue tail
 };
 class BQueueHolder
 {
 public:
 BQueueHolder();//To initialize all the members
 ~BQueueHolder();
 void AddtoOwnersQueue( BQueueContainer* );//Params - pointer to the BQueueContainer which should be passed from current thread to another thread
 BQueueContainer* GetFromForeignQueue();//To get from foreign queue
 //Transfer methods
 void TransferToCommonQueue();//Should be run within the host thread
 void TransferFromCommonQueue();//Should be run within the foreign thread
 public:
 IRSQueue * p_OwnersQueue, * p_CommonQueue, * p_ForeignQueue;
 pthread_mutex_t mut_Lock;

 };*/

/*Memory allocation and deallocation for the data item for BQueueContainer must be done externally
 * Casting or inheritance is up to the developer
 */
BQueueContainer::BQueueContainer()
{
	p_Data = NULL;
	p_Next = NULL;
}
BQueueContainer::BQueueContainer(void * _pData)
{
	p_Data = _pData;
	p_Next = NULL;
}

BQueue::BQueue()
{
	p_Head = NULL;
	p_Tail = NULL;
}
void BQueue::AddtoQueue(BQueueContainer * _pDataContainer)
{
	if (p_Head)
	{
		//If the queue is not empty
		p_Tail->p_Next = _pDataContainer;
	}
	else
	{
		//If the queue is empty
		p_Head = _pDataContainer;
	}
	p_Tail = _pDataContainer;
}

BQueueHolder::BQueueHolder()
{
	p_OwnersQueue = new BQueue();
	p_CommonQueue = new BQueue();
	p_ForeignQueue = new BQueue();
	pthread_mutex_init(&mut_Lock, NULL);

}
BQueueHolder::~BQueueHolder()
{
	delete p_OwnersQueue;
	delete p_CommonQueue;
	delete p_ForeignQueue;
}
void BQueueHolder::AddtoOwnersQueue(BQueueContainer * _pDataContainer)
{
	p_OwnersQueue->AddtoQueue(_pDataContainer);
}
void BQueueHolder::AddtoOwnersQueueWithLock(BQueueContainer * _pDataContainer)
{
	pthread_mutex_lock(&mut_Lock);
	p_OwnersQueue->AddtoQueue(_pDataContainer);
	pthread_mutex_unlock(&mut_Lock);

}
void BQueueHolder::TransferToCommonQueue()
{
	pthread_mutex_lock(&mut_Lock);
	if (p_OwnersQueue->p_Head)
	{
		if (p_CommonQueue->p_Head)
		{
			p_CommonQueue->p_Tail->p_Next = p_OwnersQueue->p_Head;
		}
		else
		{
			p_CommonQueue->p_Head = p_OwnersQueue->p_Head;
		}
		p_CommonQueue->p_Tail = p_OwnersQueue->p_Tail;
		p_OwnersQueue->p_Head = NULL;
		p_OwnersQueue->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_Lock);
}
void BQueueHolder::TransferFromCommonQueue()
{
	pthread_mutex_lock(&mut_Lock);
	if (p_CommonQueue->p_Head)
	{
		if (p_ForeignQueue->p_Head)
		{
			p_ForeignQueue->p_Tail->p_Next = p_CommonQueue->p_Head;
		}
		else
		{
			p_ForeignQueue->p_Head = p_CommonQueue->p_Head;
		}
		p_ForeignQueue->p_Tail = p_CommonQueue->p_Tail;
		p_CommonQueue->p_Head = NULL;
		p_CommonQueue->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_Lock);
}
BQueueContainer* BQueueHolder::GetFromForeignQueue()
{
	if (p_ForeignQueue->p_Head)
	{
		BQueueContainer* pCont = p_ForeignQueue->p_Head;
		p_ForeignQueue->p_Head = p_ForeignQueue->p_Head->p_Next;
		pCont->p_Next = NULL;
		return pCont;
	}
	else
		return NULL;
}

BQueueHolder2::BQueueHolder2() :
	BQueueHolder()
{
	p_OwnersPoolQ = new BQueue();
	p_CommonPoolQ = new BQueue();
	p_ForeignPoolQ = new BQueue();
	pthread_mutex_init(&mut_PoolLock, NULL);
}
BQueueHolder2::BQueueHolder2(int _iContPoolSize) :
	BQueueHolder()
{
	p_OwnersPoolQ = new BQueue();
	p_CommonPoolQ = new BQueue();
	p_ForeignPoolQ = new BQueue();
	pthread_mutex_init(&mut_PoolLock, NULL);
	i_ContPoolSize = _iContPoolSize;

	BQueueContainer * pCont;
	for (int i = 0; i < i_ContPoolSize; ++i)
	{
		pCont = new BQueueContainer(NULL);
		p_OwnersPoolQ->AddtoQueue(pCont);
	}
}
BQueueHolder2::~BQueueHolder2()
{
	BQueueContainer * pCont;
	if (p_OwnersPoolQ)
	{
		while (p_OwnersPoolQ->p_Head)
		{
			pCont = p_OwnersPoolQ->p_Head;
			p_OwnersPoolQ->p_Head = p_OwnersPoolQ->p_Head->p_Next;
			delete pCont;
		}
		delete p_OwnersPoolQ;
	}
	if (p_CommonPoolQ)
	{
		while (p_CommonPoolQ->p_Head)
		{
			pCont = p_CommonPoolQ->p_Head;
			p_CommonPoolQ->p_Head = p_CommonPoolQ->p_Head->p_Next;
			delete pCont;
		}
		delete p_CommonPoolQ;
	}
	if (p_ForeignPoolQ)
	{
		while (p_ForeignPoolQ->p_Head)
		{
			pCont = p_ForeignPoolQ->p_Head;
			p_ForeignPoolQ->p_Head = p_ForeignPoolQ->p_Head->p_Next;
			delete pCont;
		}
		delete p_ForeignPoolQ;
	}
}

void BQueueHolder2::SetPoolSize(int _iContPoolSize)
{

	i_ContPoolSize += _iContPoolSize;

	BQueueContainer * pCont;
	for (int i = 0; i < _iContPoolSize; ++i)
	{
		pCont = new BQueueContainer(NULL);
		p_OwnersPoolQ->AddtoQueue(pCont);
	}

}
void BQueueHolder2::AddtoOwnersQueue(void * _pData)
{
	BQueueContainer * pCont;
	if (p_OwnersPoolQ->p_Head)
	{
		pCont = p_OwnersPoolQ->p_Head;
		p_OwnersPoolQ->p_Head = p_OwnersPoolQ->p_Head->p_Next;
		pCont->p_Next = NULL;
		pCont->p_Data = _pData;
	}
	else
	{
		pCont = new BQueueContainer(_pData);
		i_ContPoolSize++;
	}
	p_OwnersQueue->AddtoQueue(pCont);
}
void BQueueHolder2::AddtoOwnersQueueWithLock(void * _pData)
{
	BQueueContainer * pCont;
	if (p_OwnersPoolQ->p_Head)
	{
		pCont = p_OwnersPoolQ->p_Head;
		p_OwnersPoolQ->p_Head = p_OwnersPoolQ->p_Head->p_Next;
		pCont->p_Next = NULL;
		pCont->p_Data = _pData;
	}
	else
	{
		pCont = new BQueueContainer(_pData);
		i_ContPoolSize++;
	}
	pthread_mutex_lock(&mut_Lock);
	p_OwnersQueue->AddtoQueue(pCont);
	pthread_mutex_unlock(&mut_Lock);

}
void BQueueHolder2::TransferToCommonQueue()
{
	pthread_mutex_lock(&mut_Lock);
	if (p_OwnersQueue->p_Head)
	{
		if (p_CommonQueue->p_Head)
		{
			p_CommonQueue->p_Tail->p_Next = p_OwnersQueue->p_Head;
		}
		else
		{
			p_CommonQueue->p_Head = p_OwnersQueue->p_Head;
		}
		p_CommonQueue->p_Tail = p_OwnersQueue->p_Tail;
		p_OwnersQueue->p_Head = NULL;
		p_OwnersQueue->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_Lock);

	pthread_mutex_lock(&mut_PoolLock);
	if (p_CommonPoolQ->p_Head)
	{
		if (p_OwnersPoolQ->p_Head)
		{
			p_OwnersPoolQ->p_Tail->p_Next = p_CommonPoolQ->p_Head;
		}
		else
		{
			p_OwnersPoolQ->p_Head = p_CommonPoolQ->p_Head;
		}
		p_OwnersPoolQ->p_Tail = p_CommonPoolQ->p_Tail;
		p_CommonPoolQ->p_Head = NULL;
		p_CommonPoolQ->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_PoolLock);
}
void BQueueHolder2::TransferFromCommonQueue()
{
	pthread_mutex_lock(&mut_Lock);
	if (p_CommonQueue->p_Head)
	{
		if (p_ForeignQueue->p_Head)
		{
			p_ForeignQueue->p_Tail->p_Next = p_CommonQueue->p_Head;
		}
		else
		{
			p_ForeignQueue->p_Head = p_CommonQueue->p_Head;
		}
		p_ForeignQueue->p_Tail = p_CommonQueue->p_Tail;
		p_CommonQueue->p_Head = NULL;
		p_CommonQueue->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_Lock);

	pthread_mutex_lock(&mut_PoolLock);
	if (p_ForeignPoolQ->p_Head)
	{
		if (p_CommonPoolQ->p_Head)
		{
			p_CommonPoolQ->p_Tail->p_Next = p_ForeignPoolQ->p_Head;
		}
		else
		{
			p_CommonPoolQ->p_Head = p_ForeignPoolQ->p_Head;
		}
		p_CommonPoolQ->p_Tail = p_ForeignPoolQ->p_Tail;
		p_ForeignPoolQ->p_Head = NULL;
		p_ForeignPoolQ->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_PoolLock);
}
void BQueueHolder2::AddtoForeignPool(BQueueContainer * _pCont)
{
	p_ForeignPoolQ->AddtoQueue(_pCont);
}
void BQueueHolder2::ClearPool(int _iThershold)
{
	if (i_ContPoolSize > _iThershold)
	{
		BQueueContainer * pCont;
		while (p_ForeignPoolQ->p_Head)
		{
			pCont = p_ForeignPoolQ->p_Head;
			p_ForeignPoolQ->p_Head = p_ForeignPoolQ->p_Head->p_Next;
			delete pCont;
			i_ContPoolSize--;
		}
	}
}

// ----------------------------------------------------------------------

BMultiQueueSource::BMultiQueueSource(BMultiQueueHolder * _pHolder)
{
	p_Head = NULL;
	p_Tail = NULL;
	p_Holder = _pHolder;
}

void BMultiQueueSource::Push(BQueueContainer * _pCont)
{
	if (p_Head)
	{
		//If the queue is not empty
		p_Tail->p_Next = _pCont;
	}
	else
	{
		//If the queue is empty
		p_Head = _pCont;
	}
	p_Tail = _pCont;
}

void BMultiQueueSource::Flush()
{
	//p_Holder->TransferToCommonQueue(this);

	if (p_Head)
	{
		pthread_mutex_lock(&p_Holder->mut_Lock);
		if (p_Holder->p_CommonQueue->p_Head)
		{
			p_Holder->p_CommonQueue->p_Tail->p_Next = p_Head;
		}
		else
		{
			p_Holder->p_CommonQueue->p_Head = p_Head;
		}
		p_Holder->p_CommonQueue->p_Tail = p_Tail;
		pthread_mutex_unlock(&p_Holder->mut_Lock);

		p_Head = NULL;
	}
	p_Tail = NULL;
}

void BMultiQueueSource::PushAndFlush(BQueueContainer * _pCont)
{
	if (p_Head)
	{
		p_Tail->p_Next = _pCont;
		p_Tail = _pCont;

		pthread_mutex_lock(&p_Holder->mut_Lock);

		if (p_Holder->p_CommonQueue->p_Head)
		{
			p_Holder->p_CommonQueue->p_Tail->p_Next = p_Head;
		}
		else
		{
			p_Holder->p_CommonQueue->p_Head = p_Head;
		}
		p_Holder->p_CommonQueue->p_Tail = p_Tail;
	}
	else
	{
		pthread_mutex_lock(&p_Holder->mut_Lock);

		if (p_Holder->p_CommonQueue->p_Head)
		{
			p_Holder->p_CommonQueue->p_Tail->p_Next = _pCont;
		}
		else
		{
			p_Holder->p_CommonQueue->p_Head = _pCont;
		}
		p_Holder->p_CommonQueue->p_Tail = _pCont;

	}
	pthread_mutex_unlock(&p_Holder->mut_Lock);

	p_Head = NULL;
	p_Tail = NULL;
}

BMultiQueueHolder::BMultiQueueHolder()
{
	p_CommonQueue = new BQueue();
	p_ForeignQueue = new BQueue();
	pthread_mutex_init(&mut_Lock, NULL);
	lst_SourceQueue.clear();
}

BMultiQueueHolder::~BMultiQueueHolder()
{
	delete p_CommonQueue;
	delete p_ForeignQueue;
	std::list<BMultiQueueSource*>::iterator ite = lst_SourceQueue.begin();
	while (ite != lst_SourceQueue.end())
	{
		delete (*ite);
		++ite;
	}
	lst_SourceQueue.clear();
}

BMultiQueueSource * BMultiQueueHolder::GetNewSource()
{
	BMultiQueueSource * pSource = new BMultiQueueSource(this);
	lst_SourceQueue.push_back(pSource);
	return pSource;
}

void BMultiQueueHolder::TransferToCommonQueue(BMultiQueueSource * _pSourceQueue)
{
	if (_pSourceQueue->p_Head)
	{
		pthread_mutex_lock(&mut_Lock);
		if (p_CommonQueue->p_Head)
		{
			p_CommonQueue->p_Tail->p_Next = _pSourceQueue->p_Head;
		}
		else
		{
			p_CommonQueue->p_Head = _pSourceQueue->p_Head;
		}
		p_CommonQueue->p_Tail = _pSourceQueue->p_Tail;
		pthread_mutex_unlock(&mut_Lock);

		_pSourceQueue->p_Head = NULL;
	}
	_pSourceQueue->p_Tail = NULL;
}

void BMultiQueueHolder::TransferFromCommonQueue()
{
	pthread_mutex_lock(&mut_Lock);
	if (p_CommonQueue->p_Head)
	{
		if (p_ForeignQueue->p_Head)
		{
			p_ForeignQueue->p_Tail->p_Next = p_CommonQueue->p_Head;
		}
		else
		{
			p_ForeignQueue->p_Head = p_CommonQueue->p_Head;
		}
		p_ForeignQueue->p_Tail = p_CommonQueue->p_Tail;
		p_CommonQueue->p_Head = NULL;
		p_CommonQueue->p_Tail = NULL;
	}
	pthread_mutex_unlock(&mut_Lock);
}

BQueueContainer * BMultiQueueHolder::Pop()
{
	if (p_ForeignQueue->p_Head)
	{
		BQueueContainer* pCont = p_ForeignQueue->p_Head;
		p_ForeignQueue->p_Head = p_ForeignQueue->p_Head->p_Next;
		pCont->p_Next = NULL;
		return pCont;
	}
	else
		return NULL;
}

